package cn.com.itg.function.process;

import cn.com.itg.pojo.domain.entity.ZTTCIDENTICAL;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

public class ZttDbConnectCdcFunction extends CoProcessFunction<ZTTCIDENTICAL, List<ZTTCIDENTICAL>, List<ZTTCIDENTICAL>>  implements Serializable {
    // 配置信息未加载完毕时先缓存数据流元素
    // 存储DB过来的数据
    private List<ZTTCIDENTICAL> ztts = new ArrayList<>();
//    @Override
//    public void open(Configuration parameters) throws Exception {
//        ztts = getRuntimeContext()
//                .getListState(new ListStateDescriptor<>("ztt_db", TypeInformation.of(new TypeHint<ZTTCIDENTICAL>() {})))
//        ;
//    }
    @Override
    public void processElement1(ZTTCIDENTICAL zttcidentical, Context context, Collector<List<ZTTCIDENTICAL>> collector) throws Exception {
//        List<ZTTCIDENTICAL> zttList = Lists.newArrayList(ztts.get().iterator());
        List<ZTTCIDENTICAL> update = ztts.stream().map(z -> {
                    if (zttcidentical.getOpType().equals("Update")
                            && Objects.equals(zttcidentical.getZstrc(), z.getZstrc())
                            && Objects.equals(zttcidentical.getZdtype(), z.getZdtype())
                            && Objects.equals(zttcidentical.getZfield(), z.getZfield())
                            && Objects.equals(zttcidentical.getZfield2(), z.getZfield2())) {
                        return zttcidentical;
                    }
                    return z;
                }
        ).collect(Collectors.toList());
        if(zttcidentical.getOpType().equals("Insert")){
            update.add(zttcidentical);
        }
        ztts = update;
        collector.collect(update);
    }

    @Override
    public void processElement2(List<ZTTCIDENTICAL> zttcidenticals, Context context, Collector<List<ZTTCIDENTICAL>> collector) throws Exception {
        ztts = zttcidenticals;
        collector.collect(ztts);
    }
}
